package com.at.activemq.DLQ_TEST;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.junit.Test;

import javax.jms.*;


// 测试死信队列
public class JmsConsumer {

    public static final String ACTIVEMQ_URL = "tcp://192.168.91.133:61616";
    public static final String QUEUE_NAME = "queue_redelivery";

    public static void main(String[] args) throws Exception {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setMaximumRedeliveries(3);
        //  当你第四次消费消息的时候就会将消息加入死信队列
        activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(QUEUE_NAME);
        MessageConsumer messageConsumer = session.createConsumer(queue);
        messageConsumer.setMessageListener(new MessageListener() {
            public void onMessage(Message message) {
                try {
                    if (null != message && message instanceof TextMessage) {
                        TextMessage textMessage = (TextMessage) message;
                        System.out.println("****cluster msg：" + textMessage.getText());
                    }
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });

//        System.in.read();
        messageConsumer.close();
//        session.commit();
        session.close();
        connection.close();
    }

    @Test
    public void testRedeliveryConsumer() throws JMSException {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setMaximumRedeliveries(3);
        //  当你第四次消费消息的时候就会将消息加入死信队列
        activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(QUEUE_NAME);
        MessageConsumer messageConsumer = session.createConsumer(queue);
        TextMessage message = null;
        while (true) {
            message = (TextMessage) messageConsumer.receive(1000L);
            if (message != null) {
                System.out.println("消费者收到的消息：" + message.getText());
            } else {
                break;
            }
        }
//        System.in.read();
        messageConsumer.close();
//        session.commit();
        session.close();
        connection.close();
        System.out.println("消费完成收到消息");
    }
}